package com.zx.mes.disruptor.hyl.heigh;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.*;

/**
 * disruptor instance and setting
 *
 * @author huayunlong
 * @date 2018-9-6
 */
public class Main {



    public static void main(String[] args) throws InterruptedException {
        /**
         * 如查有五个消费者，在单消费者模式下，每一个都需要独占一个线程，可看源码
         */
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

        Disruptor<TradeEvent> disruptor = new Disruptor<>(() -> new TradeEvent(),
                1024 * 1024,
                executor,
                ProducerType.SINGLE,
                new BusySpinWaitStrategy());

        /// 绑定消费者

        /// 并行
//        disruptor.handleEventsWith(new TradeHandler1(), new TradeHandler2(), new TradeHandler3());

//        EventHandlerGroup<TradeEvent> tradeEventEventHandlerGroup = disruptor.handleEventsWith(new TradeHandler1(), new TradeHandler2());
//        tradeEventEventHandlerGroup.then(new TradeHandler3());

        /// 串行
//        disruptor.handleEventsWith(new TradeHandler1())
//                .handleEventsWith(new TradeHandler2())
//                .handleEventsWith(new TradeHandler3());

        /// -------- 下面的这种写法是错误的每次都是新的对象,判断TradeHandler1的时候会找不到为空 ------------
//        disruptor.handleEventsWith(new TradeHandler1());
//        disruptor.after(new TradeHandler1()).handleEventsWith(new TradeHandler2(), new TradeHandler3());
//        disruptor.after(new TradeHandler2(), new TradeHandler3()).handleEventsWith(new TradeHandler4());

        /// -------- 下面的这种写法是错误的每次都是新的对象,判断TradeHandler1的时候会找不到为空 ------------
        TradeHandler1 handler1 = new TradeHandler1();
        TradeHandler2 handler2 = new TradeHandler2();
        TradeHandler3 handler3 = new TradeHandler3();
        TradeHandler4 handler4 = new TradeHandler4();
        TradeHandler5 handler5 = new TradeHandler5();
        disruptor.handleEventsWith(handler1);
        disruptor.after(handler1).handleEventsWith(handler2, handler3);
        disruptor.after(handler2, handler3).handleEventsWith(handler4).then(handler5);

        /// start disruptor
        disruptor.start();

        CountDownLatch countDownLatch = new CountDownLatch(1);

        /// 创建线程工厂,生产线程发送任务
        ThreadFactory threadFactory = Executors.defaultThreadFactory();

        Thread thread = threadFactory.newThread(new PublishTradeEvent(countDownLatch, disruptor));

        thread.start();

        countDownLatch.await();

        Thread.sleep(2000);

        disruptor.shutdown();
        executor.shutdown();
    }
}
